package com.mai.realtime.app.dwd.db;

import com.mai.realtime.app.BaseSqlApp;
import com.mai.realtime.common.Constant;
import com.mai.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Dwd_Paper_question_result extends BaseSqlApp {
    public static void main(String[] args) {
        new Dwd_Paper_question_result().init(3005, 2, "Dwd_Paper_question_result");
    }

    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        // 1. 读取ods_db数据
        readOdsDb(tEnv, "Dwd_Paper_question_result");




        // 2. 过滤test_exam数据

//        Table cartInfo = tEnv.sqlQuery("select " +
//                " * " +
//                "from ods_db " );

        Table test_exam_question = tEnv.sqlQuery("select " +
                "data['id'] id, " +
                "data['exam_id'] exam_id, " +
                "data['paper_id'] paper_id, " +
                "data['question_id'] question_id, " +
                "data['user_id'] user_id, " +
                "data['is_correct'] is_correct, " +
                "ts " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='test_exam_question' " +
                "and (" +
                " (`type`='bootstrap-insert' and data['id'] is not null) or `type`='insert' " +
                ")");



        tEnv.createTemporaryView("test_exam_question", test_exam_question);


//        Table test_paper = tEnv.sqlQuery("select " +
//                "data['id'] paper_id, " +
//                "data['course_id'] course_id, " +
//                "data['paper_title'] paper_title, " +
//                "ts, " +
//                "pt " +
//                "from ods_db " +
//                "where `database`='edu' " +
//                "and `table`='test_paper' " +
//                "and (" +
//                " (`type`='bootstrap-insert' and data['id'] is not null) or `type`='insert' " +
//                ")");
//
//        tEnv.createTemporaryView("test_paper", test_paper);
//
//        Table course_info = tEnv.sqlQuery("select " +
//                "data['id'] course_id, " +
//                "data['course_name'] course_name, " +
//                "ts, " +
//                "pt " +
//                "from ods_db " +
//                "where `database`='edu' " +
//                "and `table`='course_info' " +
//                "and (" +
//                " (`type`='bootstrap-insert' and data['id'] is not null) or `type`='insert' " +
//                ")");
//
//        tEnv.createTemporaryView("course_info", course_info);
//
//
//        // 4.两表join
//        Table result = tEnv.sqlQuery("select " +
//                                         " tp.paper_id, " +
//                                         " te.user_id, " +
//                                         " te.exam_id, " +
//                                         " te.score, " +
//                                         " te.duration_sec, " +
//                                         " tp.paper_title paper_title, " +
//                                         " tp.course_id, " +
//                                         " ci.course_name, " +
//                                         " te.ts " +
//                                         " from test_paper tp " +
//                                         " join test_exam te" +
//                                         " on tp.paper_id = te.paper_id " +
//                                         " join course_info ci " +
//                                         " on tp.course_id = ci.course_id");
//
//        tEnv.createTemporaryView("result", result);

        //查询


        // 5. 定义一个动态表与kafka的topic关联
        tEnv.executeSql("create table dwd_dwd_paper_question( " +
                "id string, " +
                "exam_id string, " +
                "paper_id string, " +
                "question_id string, " +
                "user_id string, " +
                "is_correct string, " +
                "ts bigint " +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_PAPER_QUESTION)
        );
//
        test_exam_question.executeInsert("dwd_dwd_paper_question");
//

    }
}